package cn.xuexiyuan.flinkstudy.sql;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description: 将 DataStream 注册为 Table 和 SQL 并进行 sum 统计
 *
 * @Author 左龙龙
 * @Date 21-3-29
 * @Version 1.0
 **/
public class Demo02 {

    public static void main(String[] args) throws Exception{
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 1.source
        DataStreamSource<Tuple2<String,Integer>> words = env.fromCollection(Arrays.asList(
                Tuple2.of("flink", 1),
                Tuple2.of("flink", 1),
                Tuple2.of("flink", 1)
        ));


        // 2.transformation
        tableEnv.createTemporaryView("t_word", words, $("name"), $("count"));
        String sql = "select name, sum(`count`) as `count` from t_word group by name";
        Table resultTable = tableEnv.sqlQuery(sql);
        resultTable.printSchema();

        // 将 Table 准换为 DataStream
        // toRetractStream -> 将计算后的新数据在 DataStream 原数据的基础上更新 true 或删除 false
        DataStream<Tuple2<Boolean, Tuple2<String,Integer>>> orderDS = tableEnv.toRetractStream(resultTable, Types.TUPLE(Types.STRING, Types.INT));

        // 3.sink
        orderDS.print("sum result: ");

        // 4.excute
        env.execute();

    }
}
